package net.xmmpp.uc.im.internal.web.controller;

import net.xmmpp.uc.im.common.model.MsgLog;
import net.xmmpp.uc.im.internal.web.config.FlinkProperties;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.client.deployment.StandaloneClusterId;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.queryablestate.client.QueryableStateClient;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Collection;
import java.util.concurrent.CompletableFuture;


@RestController
@RequestMapping("/repository")
public class RepositoryController {

    private static final Logger LOGGER = LoggerFactory.getLogger(RepositoryController.class);

    @Autowired
    private FlinkProperties flinkProperties;

    @Autowired
    private Configuration configuration;

    @Autowired
    private QueryableStateClient stateClient;

    @Autowired
    @Qualifier("msgLogListDescriptor")
    private ListStateDescriptor<MsgLog> msgLogListDescriptor;

    @Autowired
    @Qualifier("logSizeDescriptor")
    private ValueStateDescriptor<Integer> logSizeDescriptor;

    @GetMapping("/msg-logs/{uid}")
    public Iterable<MsgLog> getMsgLogs(@PathVariable("uid") String uid) throws Exception{
        JobID jobId = getJobId(configuration);

        CompletableFuture<ListState<MsgLog>> msgLogLisResultFuture =
                stateClient.getKvState(jobId, "QNMsgLogList", uid, BasicTypeInfo.STRING_TYPE_INFO, msgLogListDescriptor);

        Iterable<MsgLog> msgLogListValue = msgLogLisResultFuture.get().get();
//        msgLogListValue.forEach( e -> LOGGER.info("QNMsgLogList = " + e.toString()) );


/*
        ListStateDescriptor<MsgLog> descriptor = new ListStateDescriptor<>(
                "msgLogListSD",
                TypeInformation.of(new TypeHint<MsgLog>() {}).createSerializer(new ExecutionConfig()));
        CompletableFuture<ListState<MsgLog>> msgLogLisResultFuture =
                stateClient.getKvState(jobId, "QNMsgLogList", "snowing", BasicTypeInfo.STRING_TYPE_INFO, descriptor);

        Iterable<MsgLog> msgLogListValue = msgLogLisResultFuture.get().get();
        msgLogListValue.forEach( e -> System.out.println("QNMsgLogList = " + e.toString()) );
*/


        return msgLogListValue;
    }

    @GetMapping("/log-size/{uid}")
    public Integer getMsgSize(@PathVariable("uid") String uid) throws Exception{
        JobID jobId = getJobId(configuration);

        CompletableFuture<ValueState<Integer>> resultFuture =
                stateClient.getKvState(jobId, "QNLogSize", uid, BasicTypeInfo.STRING_TYPE_INFO, logSizeDescriptor);
        Integer value = resultFuture.get().value();
        LOGGER.info("QNLogSize = " + value);

        return value;
    }

    private JobID getJobId(Configuration config){
        RestClusterClient restClient;
        CompletableFuture<Collection<JobStatusMessage>> jobStatusMessagesFuture;
        Collection<JobStatusMessage> jobCtl;

        try {
            restClient = new RestClusterClient(config, StandaloneClusterId.getInstance());
            jobStatusMessagesFuture = restClient.listJobs();
            jobCtl = jobStatusMessagesFuture.get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        JobID jobID = null;
        for(JobStatusMessage job : jobCtl){
            if(job.getJobName().equals(flinkProperties.getJobName())){
                jobID = job.getJobId();
                break;
            }
        }

        return jobID;
    }
}
